package com.yyw.rabbitmq.core.config;

import com.google.common.base.Preconditions;
import com.yyw.rabbitmq.api.Message;
import com.yyw.rabbitmq.api.MessageType;
import com.yyw.rabbitmq.core.serialize.JacksonSerializerFactory;
import com.yyw.rabbitmq.core.serialize.conver.GenericMessageConverter;
import com.yyw.rabbitmq.core.serialize.conver.RabbitMessageConverterDecorator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RabbitMQ池
 * 根据不同的 exchange，存放不同的 RabbitmqTemplate
 */
@Slf4j
@Component
public class RabbitMQPool {
    /**
     * key: exchange 名字
     * value: rabbitTemplate实例
     */
    private Map<String, RabbitTemplate> rabbitMap = new ConcurrentHashMap<>();

    @Autowired
    private ConnectionFactory connectionFactory;

    public RabbitTemplate getRabbit(Message message) {
        Preconditions.checkNotNull(message);
        String exchange = message.getTopic();
        if (rabbitMap.get(exchange) != null) {
            return rabbitMap.get(exchange);
        }

        log.warn("exchange: {} 不存在，准备创建...", exchange);
        String routeKey = message.getRouteKey();
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setExchange(exchange);
        rabbitTemplate.setRoutingKey(routeKey);
        // 重试
        rabbitTemplate.setRetryTemplate(new RetryTemplate());
        if (!Objects.equals(message.getMessageType(), MessageType.QUICK)) {
            // confirm 回调
            rabbitTemplate.setConfirmCallback(confirmCallback);
        }
        // 自定义序列化反序列化规则,没有使用装饰者，直接使用原有的
//        rabbitTemplate.setMessageConverter(new GenericMessageConverter(new JacksonSerializerFactory().create()));
        // 使用装饰者模式
        GenericMessageConverter converter = new GenericMessageConverter(new JacksonSerializerFactory().create());
        RabbitMessageConverterDecorator decorator = new RabbitMessageConverterDecorator(converter);
        rabbitTemplate.setMessageConverter(decorator);

        rabbitMap.putIfAbsent(exchange, rabbitTemplate);
        return rabbitTemplate;
    }

    /**
     * 回调消息
     */
    private RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        String[] data = correlationData.getId().split("#");
        String id = data[0];
        String sendTime = data[1];
        log.info("消息反馈结果:[消息ID: {}, 消息发送时间: {}, 投递成功: {}, cause: {}]", id, sendTime, ack, cause);
        if (ack) {
            log.info("消息投递成功");
        } else {
            log.error("消息投递失败");
        }
    };

}
